package com.bonc.jedisCluster.utils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import com.bonc.jedisCluster.connectionHandler.MyJedisSlotBasedConnectionHandler;

import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisClusterConnectionHandler;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.exceptions.JedisAskDataException;
import redis.clients.jedis.util.JedisClusterCRC16;

/**
 * 能够处理cluster模式的slots迁移的问题
 * @author WYB
 *
 */
public class PipelineHandler {
	private Jedis jedis;
	private JedisClusterConnectionHandler connectionHandler;
	
	public PipelineHandler(JedisClusterConnectionHandler connectionHandler) {
		this.connectionHandler = connectionHandler;
	}

	public PipelineHandler(Jedis jedis, JedisClusterConnectionHandler connectionHandler) {
		this.jedis = jedis;
		this.connectionHandler = connectionHandler;
	}
	
	/**
	 * @param keys 映射到当前jedis节点的所有keys，不支持跨节点
	 * @return
	 */
	public List<String> mget(String... keys) {
		List<String> results = new ArrayList<String>();
		List<String> keyList = Arrays.asList(keys);
		if (this.jedis==null) {
			this.jedis= ((MyJedisSlotBasedConnectionHandler) this.connectionHandler).getConnectionFromSlot(JedisClusterCRC16.getSlot(keyList.get(0)));
		}
		Pipeline pipelined = this.jedis.pipelined();
		for (String key: keyList) {
			pipelined.get(key);
		}
		List<Object> returns = pipelined.syncAndReturnAll();
		for (int i = 0; i < keyList.size(); i++) {
			Object ret = returns.get(i);
			if (ret !=null) {
				// 如果有哪个key的执行结果是AskException，说明这个key已经迁移了
				if (ret instanceof JedisAskDataException) {
					JedisAskDataException askException = (JedisAskDataException) ret;
					HostAndPort targetNode = askException.getTargetNode();
					Jedis targetJedis = this.connectionHandler.getConnectionFromNode(targetNode);
					String targetResult = null;
					try {
						// 调用asking接口，这样就可以在targetJedis上查询到这个key的值
						targetJedis.asking();
						// 由于pipeline处理结果是有序的，才可以这样来得到key
						String key = keyList.get(i);
						targetResult = targetJedis.get(key);
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						releaseConnection(targetJedis);
					}
					results.add(targetResult);
				} else {
					results.add((String) ret);
				}
			} else {
				results.add(null);
			}
		}
		return results;
	}
	
	public List<String> mset(String... keysvalues) {
		List<String> results = new ArrayList<String>();
		String[] keys = new String[keysvalues.length / 2];
		String[] values = new String[keysvalues.length / 2];
		for (int keyIdx = 0; keyIdx < keys.length; keyIdx++) {
		      keys[keyIdx] = keysvalues[keyIdx * 2];
		      values[keyIdx] = keysvalues[keyIdx * 2 +1];
		    }
		List<String> keyList = Arrays.asList(keys);
		List<String> valueList = Arrays.asList(values);
		if (this.jedis==null) {
			this.jedis= ((MyJedisSlotBasedConnectionHandler) this.connectionHandler).getConnectionFromSlot(JedisClusterCRC16.getSlot(keyList.get(0)));
		}
		Pipeline pipelined = this.jedis.pipelined();
		for (int i = 0; i < keyList.size(); i++) {
			pipelined.set(keyList.get(i), valueList.get(i));
		}
		List<Object> returns = pipelined.syncAndReturnAll();
		for (int i = 0; i < keyList.size(); i++) {
			Object ret = returns.get(i);
			if (ret !=null) {
				// 如果有哪个key的执行结果是AskException，说明这个key已经迁移了
				if (ret instanceof JedisAskDataException) {
					JedisAskDataException askException = (JedisAskDataException) ret;
					HostAndPort targetNode = askException.getTargetNode();
					Jedis targetJedis = this.connectionHandler.getConnectionFromNode(targetNode);
					String targetResult = null;
					try {
						// 调用asking接口，这样就可以在targetJedis上查询到这个key的值
						targetJedis.asking();
						String key = keyList.get(i);
						String value = valueList.get(i);
						targetResult = targetJedis.set(key, value);
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						releaseConnection(targetJedis);
					}
					results.add(targetResult);
				} else {
					results.add((String) ret);
				}
			} else {
				results.add(null);
			}
		}
		return results;
	}
	
	private void releaseConnection(Jedis connection) {
		if (connection != null) {
			connection.close();
		}
	}
}
